/*
 * Copyright 2014 Semidata. All rights reserved.
 */
package com.semidata.trp.report.sort;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.Comparator;
import java.util.List;
import java.util.Random;

/**
 * @author shewei@semidata.com, She Wei
 * 
 */
public class MergeSort<T> {

	private List<DataProvider<T>> providers;
	private int[] losers;
	private Comparator<T> comparator;

	public MergeSort(List<DataProvider<T>> providers, Comparator<T> comparator) {

		if (providers == null || providers.isEmpty()) {
			throw new RuntimeException("no data provider");
		}

		if (comparator == null) {
			throw new RuntimeException("no comparator");
		}

		this.providers = providers;
		this.comparator = comparator;

		init();
	}

	private void init() {

		int size = providers.size();

		int[] winners = new int[size * 2];
		losers = new int[size * 2];

		for (int i = size; i < size * 2; i++) {
			winners[i] = i - size;
			losers[i] = i - size;
		}

		for (int i = size - 1; i > 0; i--) {
			int left = winners[2 * i];
			int right = winners[2 * i + 1];

			if (win(providers.get(left).peek(), providers.get(right).peek())) {
				winners[i] = left;
				losers[i] = right;
			} else {
				winners[i] = right;
				losers[i] = left;
			}
		}

		losers[0] = winners[1];

	}

	private boolean win(T obj1, T obj2) {

		if (obj1 == null && obj2 == null) {
			return true;
		} else if (obj1 == null) {
			return false;
		} else if (obj2 == null) {
			return true;
		} else {
			return comparator.compare(obj1, obj2) <= 0;
		}

	}

	private void adjust() {
		int index = losers[0] + providers.size();
		int winner = losers[0];

		while (index > 1) {
			int parent = index / 2;

			if (!win(providers.get(winner).peek(), providers
			        .get(losers[parent]).peek())) {
				int tmp = winner;
				winner = losers[parent];
				losers[parent] = tmp;
			}
			index = parent;
		}

		losers[0] = winner;
	}

	public T poll() {
		T obj = providers.get(losers[0]).poll();
		adjust();
		return obj;
	}

	public T peek() {
		return providers.get(losers[0]).peek();
	}

	private static class TestProvider implements DataProvider<Integer> {

		private int[] data = null;
		private int index = 0;
		private int size = 0;

		public TestProvider(int size) {
			this.data = new int[size];
			this.size = size;
			Random rand = new Random();
			for (int i = 0; i < size; i++) {
				data[i] = rand.nextInt();
			}

			Arrays.sort(data);
		}

		@Override
		public Integer peek() {
			if (index >= size)
				return null;
			return data[index];
		}

		@Override
		public Integer poll() {
			if (index >= size)
				return null;
			int ret = data[index];
			index++;
			return ret;
		}

	}

	public static void main(String[] args) {

		
		List<DataProvider<Integer>> providers = new ArrayList<>();
		for (int i = 0; i < 100; i++) {
			providers.add(new TestProvider(100000));
		}

		long start = System.currentTimeMillis();
		
		MergeSort<Integer> sorter = new MergeSort<Integer>(providers, new Comparator<Integer>() {

			@Override
            public int compare(Integer o1, Integer o2) {
	            return o1.compareTo(o2);
            }
			
		});
		
		
		List<Integer> results = new ArrayList<>();
		Integer obj;
		while ((obj = sorter.poll()) != null) {
			results.add(obj);
		}
		System.out.println((System.currentTimeMillis() - start) + " ms");
		System.out.println(results.size());
		
		boolean flag = true;
		
		for (int i = 1; i < results.size(); i++) {
			if (results.get(i - 1) > results.get(i)) {
				flag = false;
				break;
			}
		}
		
		System.out.println(flag ? "OK" : "ERROR");
		
	}

}
